package com.gy.hadoop.hbase;

import org.apache.commons.io.filefilter.PrefixFileFilter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class MainCURD {

    private Configuration conf = null;
    private Connection conn = null;
    private Admin admin = null;

    private String tableName = "t_user_info";


    @Before
    public void before() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "centos102:2181,centos102:2181,centos103:2181");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }

    @After
    public void after() throws IOException {
        admin.close();
        conn.close();
    }


    /**
     * create 't_user_info','base_info','extra_info'
     * list 查看所有表
     */
    @Test
    public void createTable() throws IOException {
        HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName.getBytes()));
        HColumnDescriptor cl1 = new HColumnDescriptor("base_info".getBytes());
        HColumnDescriptor cl2 = new HColumnDescriptor("extra_info".getBytes());
        cl2.setVersions(1, 3);

        descriptor.addFamily(cl1).addFamily(cl2);

        admin.createTable(descriptor);
    }

    /**
     * put 't_user_info','user001','base_info:useranme','zhangsan'
     * put 't_user_info','user001','base_info:password','123456'
     */
    @Test
    public void put() throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName.getBytes()));

        ArrayList<Put> puts = new ArrayList<>();

        Put put01 = new Put("user001".getBytes());
        put01.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("useranme"), "zhangsan".getBytes());

        Put put02 = new Put("user001".getBytes());
        put02.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("password"), "123456".getBytes());

        Put put03 = new Put("user001".getBytes());
        put03.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("password"), "ssadfsa".getBytes());
        put03.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("married"), "false".getBytes());

        Put put04 = new Put("zhang_zh_01".getBytes());
        put04.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("tadfaname"), "aaaa".getBytes());
        put04.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("married"), "true".getBytes());

        Put put05 = new Put("zhang_zh_01".getBytes());
        put05.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("tadfaname"), "bbbb".getBytes());
        put05.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("married"), "false".getBytes());
        put05.setDurability(Durability.ASYNC_WAL);

        puts.add(put01);
        puts.add(put02);
        puts.add(put03);
        puts.add(put04);
        puts.add(put05);

        for (int i = 0; i < 1000; i++) {
            Put put = new Put(("zhangSan_zh_00" + i).getBytes());
            put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("username"), ("zhangsan_00" + i).getBytes());
            put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("password"), ("pwd_zh_00" + i).getBytes());
            put.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("married"), ("false:zh_00" + i).getBytes());
            put.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("address"), ("beijing:zh_00" + i).getBytes());
            puts.add(put);
        }

        table.put(puts);

    }

    private void print(Result result) {
        List<Cell> cells = result.listCells();
        cells.forEach(x -> {
            byte[] row = CellUtil.cloneRow(x);
            byte[] family = CellUtil.cloneFamily(x);
            byte[] qualifier = CellUtil.cloneQualifier(x);
            byte[] value = CellUtil.cloneValue(x);
             System.out.println(new String(row) + "  column=" + new String(family) + ":" + new String(qualifier) + "   value=" + new String(value));

        });
    }


    /**
     * get 't_user_info','user001'
     */
    @Test
    public void get() throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName.getBytes()));

        Get get = new Get("user001".getBytes());
        Result result = table.get(get);
        print(result);
    }

    /**
     * scan 't_user_info'
     */
    @Test
    public void scan() throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName.getBytes()));

        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        Iterator<Result> it = scanner.iterator();
        while (it.hasNext()) {
            print(it.next());
        }
    }

    /**
     * delete 't_user_info','user001'
     */
    @Test
    public void delete() throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName.getBytes()));

        Delete del = new Delete("user001".getBytes());
        del.addColumn("base_info".getBytes(), "password".getBytes());

        table.delete(del);
    }


    private void testScan(Filter filter) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName.getBytes()));

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> it = scanner.iterator();
        while (it.hasNext()) {
            print(it.next());
        }
    }


    /**
     * rowKey前缀过滤
     *
     * @throws IOException
     */
    @Test
    public void prefixFilter() throws IOException {
        PrefixFilter filter = new PrefixFilter("zh".getBytes());
        testScan(filter);
    }


    /**
     * 行比较器 需要比较运算符
     *
     * @throws IOException
     */
    @Test
    public void rowFilter() throws IOException {
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.LESS,
                new BinaryComparator((Bytes.toBytes("user002"))));
        testScan(filter);

        System.out.println("====================================");
        RowFilter filter2 = new RowFilter(CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("zh"));
        testScan(filter2);
    }

    /**
     * 列名过滤器
     *
     * @throws IOException
     */
    @Test
    public void qualifierFilter() throws IOException {
        QualifierFilter qf1 = new QualifierFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryComparator(Bytes.toBytes("married")));
        testScan(qf1);
        System.out.println("====================================");

        QualifierFilter qf2 = new QualifierFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryPrefixComparator(Bytes.toBytes("user")));
        testScan(qf2);
    }


    /**
     * 列值过滤器
     *
     * @throws IOException
     */
    @Test
    public void valueFilter() throws IOException {
        RegexStringComparator regexStringComparator = new RegexStringComparator("^zhang");
        SubstringComparator s2 = new SubstringComparator("false");
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("base_info".getBytes(), "married".getBytes(),
                CompareFilter.CompareOp.EQUAL, s2);

        testScan(singleColumnValueFilter);
    }


    /**
     * 组合过滤器
     *
     * @throws IOException
     */
    @Test
    public void multiFilter() throws IOException {
        FamilyFilter f1 = new FamilyFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryPrefixComparator("base".getBytes()));

        ColumnPrefixFilter f2 = new ColumnPrefixFilter("password".getBytes());

        //TODO 多条件同时需要满足
        FilterList fs = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        fs.addFilter(f1);
        fs.addFilter(f2);

        testScan(fs);
    }

}
